package cc.shacocloud.redis.study.queue.delay;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import cc.shacocloud.redis.study.util.AbstractTimingTask;
import cc.shacocloud.redis.study.util.codeBlock.CodeBlockMD;
import cc.shacocloud.redis.study.util.codeBlock.CodeBlockMDManage;
import cc.shacocloud.redis.study.util.env.YamlConfigurer;
import cc.shacocloud.redis.study.util.redis.RedisKeys;
import cc.shacocloud.redis.study.util.redis.RedisUtil;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.Nullable;
import redis.clients.jedis.Builder;

import java.util.List;

/**
 * 基于redis实现延时队列
 * <p>
 * Sorted Set + Hash，利用Sorted Set高效的排序特性，对过期时间进行排序，具体数据存放在Hash中，利用全局消息id进行对应
 * <p>
 * 初始化配置：
 * <pre>
 *     redis:
 *       delay-message:
 *         prefix: rdm   # redis配置前缀
 *         startup-delay: 0 # 启动延迟
 *         time-interval: 500 # 执行间隔
 *         thread-name: redis-delay # 执行线程名称
 * </pre>
 */
@Slf4j
@CodeBlockMD(mdDirPath = "/redis", mdFileName = "delayQueue")
public abstract class RedisDelayMessage extends AbstractTimingTask implements DelayMessage, DelayMessageConsumer {

    private static final Class<RedisDelayMessage> CLASS = RedisDelayMessage.class;

    /* Sorted Set队列键 */
    private final String SORTED_SET_KEY;
    /* Hash队列键 */
    private final String HASH_KEY;

    public static final String DEFAULT_THREAD_NAME = "redis-delay";
    public static final String DEFAULT_DELAY_QUEUE_PREFIX = "rdm";

    /**
     * @param businessName 业务名称，会加入redis的key中用于区分，尽量简短
     */
    public RedisDelayMessage(String businessName) {
        this(businessName, DEFAULT_THREAD_NAME);
    }

    /**
     * @param businessName 业务名称，会加入redis的key中用于区分，尽量简短
     * @param name         线程名， 如果为空会从配置中读取，如果配置中也未设置将使用默认名称
     */
    public RedisDelayMessage(String businessName, String name) {
        super(StrUtil.blankToDefault(
                name,
                YamlConfigurer.getDefault().getStrProperty("redis.delay-message.thread-name", DEFAULT_DELAY_QUEUE_PREFIX))
        );

        YamlConfigurer configurer = YamlConfigurer.getDefault();
        RedisKeys redisKeys = new RedisKeys(
                configurer.getStrProperty("redis.delay-message.prefix", DEFAULT_DELAY_QUEUE_PREFIX)
        );

        this.SORTED_SET_KEY = redisKeys.keys(businessName, "zs");
        this.HASH_KEY = redisKeys.keys(businessName, "h");

        // 启动
        fixedDelay(
                configurer.getLongProperty("redis.delay-message.startup-delay", 0L),
                configurer.getLongProperty("redis.delay-message.time-interval", 500L)
        );
    }


    @Override
    public long delay(byte[] msg, long delay) {
        // 消息id 全局唯一
        final String msgId = IdUtil.objectId();
        // 过期时间戳 毫秒级; 保存时使用负数，方便redis Sorted Set，从高到低排序
        final long expire = DateUtil.current(false) + delay;

        final String script = CodeBlockMDManage.get(CLASS, "delay");
        RedisUtil.eval(script, 2, this.SORTED_SET_KEY, this.HASH_KEY, msgId, String.valueOf(-expire), msg);
        return expire;
    }

    @Override
    public long getStackedQuantity() {
        return RedisUtil.zcard(this.SORTED_SET_KEY);
    }

    /**
     * 消息列表数据构建器
     */
    private final Builder<DelayMsg> DELAY_MSG = new Builder<DelayMsg>() {

        @Nullable
        @Override
        @SuppressWarnings("unchecked")
        public DelayMsg build(Object data) {
            if (data == null) return null;

            List<Object> fl = (List<Object>) data;
            long expire = (long) fl.get(0);

            DelayMsg msg = new DelayMsg();
            msg.setExpire(Math.abs(expire));
            msg.setData((byte[]) fl.get(1));
            return msg;
        }
    };

    /**
     * 消费消息
     */
    protected DelayMsg consumeDelayMsg() {
        final long current = -DateUtil.current(false);
        final String script = CodeBlockMDManage.get(CLASS, "consumeDelayMsg");
        return RedisUtil.eval(this.DELAY_MSG, script, 2, this.SORTED_SET_KEY, this.HASH_KEY, String.valueOf(current));
    }

    @Override
    public final void exec() {
        if (!beforeConsume()) return;

        do {
            final DelayMsg msg = consumeDelayMsg();
            if (msg == null) break;

            final byte[] data = msg.getData();
            final long expire = msg.getExpire();
            try {
                consume(expire, data);
            } catch (Throwable cause) {
                exceptionCaught(expire, data, cause);
            }
        } while (hasNext());
    }

    /**
     * @return 消费之前函数，如果返回false则跳过本次消费
     */
    protected boolean beforeConsume() {
        return true;
    }

    /**
     * @return 继续消费数据，如果返回false 则停止本次消费
     */
    protected boolean hasNext() {
        return true;
    }


    @Setter
    @Getter
    @NoArgsConstructor
    private static class DelayMsg {

        /**
         * 消息过期时间
         */
        private long expire;

        /**
         * 消息数据
         */
        private byte[] data;

    }
}
